package com.demo.java.disruptor.netty.server;

import com.demo.java.disruptor.netty.common.AbstractConsumer;
import com.demo.java.disruptor.netty.common.DisruptorFactory;
import com.demo.java.disruptor.netty.common.MarshallingCodeCFactory;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

public class NettyServer {
    public NettyServer(ProducerType producerType, int bufferSize, WaitStrategy waitStrategy, int consumerSize) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workGroup)
                .channel(NioServerSocketChannel.class)
                // 连接队列数
                .option(ChannelOption.SO_BACKLOG, 1024*1024)
                // 接收缓冲大小
                .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel channel) throws Exception {
                        channel.pipeline()
//                                .addLast(new MessageDecode())
//                                .addLast(new MessageEncode())
                                .addLast(MarshallingCodeCFactory.buildMarshallingDecoder())
                                .addLast(MarshallingCodeCFactory.buildMarshallingEncoder())
                                .addLast(new ServerMessageHandler());
                    }
                });
        try {
            ChannelFuture channelFuture = bootstrap.bind(9000).sync();
            System.out.println("server start...");

            AbstractConsumer[] consumers = new ServerConsumer[consumerSize];
            for (int i = 0; i < consumerSize; i++) {
                consumers[i] = DisruptorFactory.createServerConsumer();
            }
            DisruptorFactory.getInstance().init(producerType, bufferSize, waitStrategy, consumers);

            channelFuture.channel().closeFuture().sync();
            System.out.println("server end...");
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}
